package chapter07

import java.util

import org.apache.kudu.{ColumnSchema, Schema, Type}
import org.apache.kudu.client.{AlterTableOptions, CreateTableOptions, Insert, KuduClient, KuduScanner, KuduSession, KuduTable, PartialRow, RowResult, RowResultIterator, Update}

import scala.collection.mutable

object KuduAPIApp {


  def main(args: Array[String]): Unit = {

    val KUDU_MASTER = "hadoop000"
    val kuduClient: KuduClient = new KuduClient.KuduClientBuilder(KUDU_MASTER).build()

    val tableName = "ods"

    //创建表
    //createTable(kuduClient, tableName)

    // insertRow(kuduClient, tableName)
    //deleteTable(kuduClient,tableName)
    deleteTable(kuduClient,"ods")
//    deleteTable(kuduClient,"provincename_city_stat_20200801")
//    deleteTable(kuduClient,"app_stat_20200801")
//    deleteTable(kuduClient,"area_stat_20200801")


    //query(kuduClient, tableName)

    //alterRow(kuduClient,tableName)

//    val newTableName = "pk_kudu"
//    renameTable(kuduClient,tableName,newTableName)
    kuduClient.close()


  }

  /**
   * 修改表名
   * @param kuduClient
   * @param tableName
   * @return
   */
  def renameTable(kuduClient: KuduClient, tableName: String,newTableName:String) = {
    //alterTable(String name, AlterTableOptions ato)
    val options = new AlterTableOptions()
    options.renameTable(newTableName)
    kuduClient.alterTable(tableName,options)
    kuduClient.close()
  }
  /*
  修改表数据
   */
  def alterRow(kuduClient: KuduClient, tableName: String) = {

    val table: KuduTable = kuduClient.openTable(tableName)
    val session: KuduSession = kuduClient.newSession()
    val update: Update = table.newUpdate()
    val row: PartialRow = update.getRow
    row.addString("word","pk-10")
    row.addInt("cnt",8888)

    session.apply(update)
    session.close()

  }



  /**
   * 查询数据
   *
   * @param kuduClient
   * @param tableName
   * @return
   */

  def query(kuduClient: KuduClient, tableName: String) = {
    val table: KuduTable = kuduClient.openTable(tableName)
    val scanner: KuduScanner = kuduClient.newScannerBuilder(table).build()
    while (scanner.hasMoreRows) {
      val iterator: RowResultIterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result: RowResult = iterator.next()
        println(result.getString("word") + "==>" + result.getInt("cnt"))
      }
    }
  }


  /**
   * 删除表
   *
   * @param kuduClient
   * @param tableName
   */
  def deleteTable(kuduClient: KuduClient, tableName: String) = {
    kuduClient.deleteTable(tableName)
  }

  /**
   * 插入数据
   * #找资料批量插入
   */
  def insertRow(client: KuduClient, tableName: String): Unit = {
    val table: KuduTable = client.openTable(tableName)
    val session: KuduSession = client.newSession()

    for (i <- 1 to 10) {
      val insert: Insert = table.newInsert()
      val row: PartialRow = insert.getRow
      row.addString("word", s"pk-$i")
      row.addInt("cnt", 100 + i)

      session.apply(insert)
    }


  }

  /**
   * 创建表
   */
  def createTable(client: KuduClient, tableName: String): Unit = {

    //createTable(String name, Schema schema, CreateTableOptions builder)
    import scala.collection.JavaConverters._
    val columns = List(

      //ColumnSchemaBuilder(String name, Type type)
      new ColumnSchema
      .ColumnSchemaBuilder("word", Type.STRING).key(true).build(),
      new ColumnSchema
      .ColumnSchemaBuilder("cnt", Type.INT32)
        //.nullable(true)//可以为空
        .build()
    ).asJava

    //Schema(List<ColumnSchema> columns)
    val schema = new Schema(columns)

    val options = new CreateTableOptions()
    options.setNumReplicas(1)
    val parcols = new util.LinkedList[String]()
    parcols.add("word")
    //ddHashPartitions(List<String> columns, int buckets)
    options.addHashPartitions(parcols, 3)
    client.createTable(tableName, schema, options)

  }
}
